package com.corn.flink.lesson5;

import com.corn.flink.lesson4.Event;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.EventTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.assigners.GlobalWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

import java.time.Duration;

/**
 * window demo
 *
 * @author JimWu
 * @date 2023/3/13 17:58
 **/
public class WindowDemo {

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(1);
        DataStreamSource<Event> ds = env.fromElements(
                new Event("add", 1000),
                new Event("delete", 2000),
                new Event("update", 3000),
                new Event("select", 5000),
                new Event("add", 6000),
                new Event("delete", 7000),
                new Event("select", 8000),
                new Event("select", 9000),
                new Event("select", 10000),
                new Event("select", 11000)
        );

        ds.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ZERO)
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                }))
                .keyBy(Event::getType)
                .window(TumblingEventTimeWindows.of(Time.seconds(10),Time.seconds(1)));//滚动窗口 偏移1S执行 从00:00:01 - > 00:00:11为一个窗口
//                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))// 滑动窗口 窗口大小10s slide为5s
//                .window(EventTimeSessionWindows.withGap(Time.seconds(3)))// session窗口 GAP 为3s
//                .window(GlobalWindows.create())// 全局窗口
    }

}
